
package cfar.utils;

import cfar.actors.Node;
import cfar.ff.data.Assocs.Assoc2;
import cfar.ff.data.List;
import cfar.ff.data.Maybe;
import cfar.ff.data.Stream;
import cfar.ff.fun.F1;

import cfar.ff.interfaces.Sequence;
import static cfar.ff.data.Maybe.*;
import static cfar.ff.fun.Seq.*;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.EOFException;
import java.io.File;
import java.io.FileFilter;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.util.logging.Level;
import java.util.logging.Logger;

import java.io.FileInputStream;
import java.net.InetAddress;
import java.util.Random;

/**
 *
 * @author badi
 */
public class IO {

    /**
     * Write an object to the output stream using the ObjectOutputStream.
     * @param o the object to write
     * @param s the stream to write to
     * @throws IOException
     */
    public static void write_object (Object o, OutputStream s) throws IOException{
        ObjectOutputStream oos = new ObjectOutputStream(new BufferedOutputStream(s));
        oos.writeObject(o);
        oos.flush();
    }


    /**
     * Read an object from the stream.
     * @param s the stream to read from
     * @return the object
     * @throws IOException
     * @throws ClassNotFoundException
     */
    public static Object read_object (InputStream s) throws IOException, ClassNotFoundException{
        ObjectInputStream ois = new ObjectInputStream(new BufferedInputStream(s));
        return ois.readObject();
    }


    public static void log(Level lvl, Class c, String msg){
        Logger.getLogger(c.getName()).log(lvl, "|" + c.getName() + "| => " + msg);
    }


    /**
     * 
     * @param synch_dir the directory InetAddresses are serialized to
     * @return stream of (InetAddresse, Port)
     */
    public static Sequence<List, Assoc2<InetAddress, Integer>> available_hosts
                        (String synch_dir, final String suffix)
    {

        F1<Object, Maybe<Assoc2<InetAddress, Integer>>>
                mkInet =
                new F1<Object, Maybe<Assoc2<InetAddress, Integer>>>() {

            public Maybe<Assoc2<InetAddress, Integer>> f(Object a) {
                    if(a == null)
                        return nothing();
                    else return just( (Assoc2<InetAddress, Integer>) a );
            }
        };

        F1<Maybe<Assoc2<InetAddress, Integer>>, Boolean>
                onlyJusts =
                new F1<Maybe<Assoc2<InetAddress, Integer>>, Boolean>() {

            public Boolean f(Maybe<Assoc2<InetAddress, Integer>> m) {
                return m.isJust;
            }
        };

        F1<Maybe<Assoc2<InetAddress, Integer>>, Assoc2<InetAddress, Integer>>
                fromJusts =
                new F1<Maybe<Assoc2<InetAddress, Integer>>, Assoc2<InetAddress, Integer>>() {

            public Assoc2<InetAddress, Integer> f(Maybe<Assoc2<InetAddress, Integer>> m) {
                return m.fromJust();
            }
        };

        return
              shuffle(
              map    (fromJusts
            , filter (onlyJusts
            , map    (mkInet
                     , wait_for_full_writes(synch_dir, suffix, Integer.MAX_VALUE, new Random().nextInt(5000))))));
    }

    /**
     * When dealing with a distributed FS like NFS on dscratch or AFS, the serialization of
     * the actor location info to synch_dir sometimes causes an empty file to be seen by the
     * current worker. This will attempt to read the object several times, sleeping between
     * each try.
     * @param synch_dir
     * @param suffix
     * @param nattempts
     * @param sleep_time
     * @return
     */
    public static Sequence<List, Object> wait_for_full_writes
            (final String synch_dir, final String suffix, final int nattempts, final long sleep_time)
    {
        List<File> files = List.list(new File(synch_dir).listFiles(new FileFilter() {

            public boolean accept(File pathname) {
                return pathname.getAbsolutePath().endsWith(suffix);
            }
        }));

        F1<File, Object> wait_for_objs = new F1<File, Object>() {

            public Object f(File f) {
                Object o = null;
                int attempt = 0;
                while(++attempt < nattempts){
                    try {
                        o = read_object(new FileInputStream(f));
                        break;
                    } catch (EOFException ex){
                        // The FS probably hasn't propagated the changes yet
                        try {
                            Thread.sleep(sleep_time);
                        } catch (InterruptedException ex1) {
                            Logger.getLogger(IO.class.getName()).log(Level.SEVERE, null, ex1);
                        }
                    } catch (IOException ex) {
                        Logger.getLogger(IO.class.getName()).log(Level.SEVERE, null, ex);
                    } catch (ClassNotFoundException ex) {
                        Logger.getLogger(IO.class.getName()).log(Level.SEVERE, null, ex);
                    }
                }
                return o;
            }
        };

        return map(wait_for_objs, files);
    }
}
